package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import com.pig4cloud.pigx.contract.vo.Contract24hTradeVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class SymbolRiseAndFall extends RankingAbstractTopic {
	private final MongoTemplate mongo;

	protected String TOPICNAME = "riseAndFall";

	public SymbolRiseAndFall(MongoTemplate mongo) {
		this.mongo = mongo;
		init();
	}

	private void init() {
		wsPush();
	}

	private void wsPush() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				// ----------该推送方案 仅限于推送信息与用户信息无关联的场景----------
				// ----------实现方式为 将所有session根据查询类型进行划分,大幅度减少查询消耗,可承受大面积用户同时访问----------
				// ----------在服务器允许的情况下 可以将每种数据类型都new一个线程去执行,可大幅度提升并发访问的推送效率----------
				// 发送消息容器
				Map<String, List<WebSocketSession>> map = new HashMap<>();
				// 用户原生请求容器
				Map<String, List<String>> paramsMap = new HashMap<>();
				// 统计所有订阅用户的订阅类型
				for (Map.Entry<WebSocketSession, List<AbstractTopicVO>> entry : subList.entrySet()) {
					if (!entry.getKey().isOpen()) {
						subList.remove(entry.getKey());
						log.info("ws 移除已关闭的session");
						continue;
					}

					// 遍历请求参数
					List<AbstractTopicVO> paramList = entry.getValue();
					paramList.forEach(param -> {
						// 请求参数解析
						List<String> dataList = param.getDataList();
						if (dataList.size() < 2) {
							return;
						}

						// 查询大写币对名称为key
						String key = dataList.get(1).toUpperCase();
						// 如果已有该类型,则在该类型中添加发送用户
						if (map.containsKey(key)) {
							map.get(key).add(entry.getKey());
						} else {
							List<WebSocketSession> sessionList = new ArrayList<>();
							sessionList.add(entry.getKey());
							map.put(key, sessionList);
							// 保存用户原生请求
							paramsMap.put(key, dataList);
						}
					});

					// 遍历所有需要发送的数据类型,将各类型数据分发给订阅的用户
					map.forEach((key, value) -> {
						// 查询数据
						Query query = new Query();
						query.addCriteria(Criteria.where("_id").is(key));
						Contract24hTradeVo mongoOne = mongo.findOne(query, Contract24hTradeVo.class);

						// 分发消息
						value.forEach(v -> {
							try {
								v.sendMessage(getTextMessage(paramsMap.get(key), mongoOne));
							} catch (IOException e) {
								e.printStackTrace();
								log.info("push rise and fall err:" + e.getMessage());
							}
						});

					});
				}
			} catch (Exception e) {
				e.printStackTrace();
				log.info("push rise and fall err:" + e.getMessage());
			}

		}, 0, 1000, TimeUnit.MILLISECONDS);
	}
}
